package com.shujia.flink.core;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

public class Demo04EventTime {
    public static void main(String[] args) throws Exception {
        // 事件时间：数据本身自带的时间
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 并行度
        env.setParallelism(1);

        /*
        数据格式：单词,时间戳（很大的整数，Long类型）
         a,1722233813000
         a,1722233814000
         a,1722233815000
         a,1722233816000
         a,1722233817000
         a,1722233818000
         a,1722233819000
         a,1722233820000
         a,1722233822000
         a,1722233827000
         */
        DataStreamSource<String> wordTsDS = env.socketTextStream("master", 8888);

        SingleOutputStreamOperator<Tuple2<String, Long>> mapDS = wordTsDS
                .map(line -> Tuple2.of(line.split(",")[0], Long.parseLong(line.split(",")[1])), Types.TUPLE(Types.STRING, Types.LONG));

        // 指定数据的时间戳，告诉Flink，将其作为事件时间进行处理
        SingleOutputStreamOperator<Tuple2<String, Long>> assDS = mapDS
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
//                                // 单调递增时间戳策略，不考虑数据乱序问题
//                                .<Tuple2<String, Long>>forMonotonousTimestamps()
                                // 容忍5s的数据乱序到达，本质上将水位线前移5s，缺点：导致任务延时变大
                                // 水位线：某个线程中所接收到的数据中最大的时间戳
                                .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                // 可以提取数据的某一部分作为事件时间
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple2<String, Long> t2, long recordTimestamp) {
                                        return t2.f1;
                                    }
                                })
                );

        // 不管是事件时间还是处理时间都需要搭配窗口操作一起使用
        assDS.map(kv -> Tuple2.of(kv.f0, 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t2 -> t2.f0)
                // 窗口触发的条件：水位线超过了窗口的结束时间
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum(1)
                .print();

        env.execute();


    }
}
